package com.lemon.core.conf.core;


import com.lemon.core.conf.BaseConfig;
import com.lemon.core.conf.ZKClient;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.io.AbstractResource;

import java.io.*;
import java.net.URL;
import java.util.Arrays;

@SuppressWarnings("unused")
public class ZookeeperResource extends AbstractResource implements ApplicationContextAware, DisposableBean {
    // 配置的zk根
    private final static String ZKCONFIGS_CHROOT = BaseConfig.getValue("zk.config.prefix", "/mop");
    private static final String URL_HEADER = "zk://";
//    static Logger log = LoggerFactory.getLogger(ZookeeperResource.class);

    public static enum ReloadContext {
        AUTO, HOLD
    }

    ;

    public static enum OnConnectionFailed {
        IGNORE, THROW_EXCEPTION
    }

    ;

    public static enum PingCmd {
        get, ls
    }

    AbstractApplicationContext ctx;

    private String zkNodePaths;

    public String getZkNodePaths() {
        return zkNodePaths;
    }

    public void setZkNodePaths(String zkNodePaths) {
        if (StringUtils.isBlank(zkNodePaths)) {
            throw new IllegalArgumentException("property zkNodePaths cannot be null");
        }
        if (zkNodePaths.startsWith(ZKCONFIGS_CHROOT) || zkNodePaths.startsWith("/"))
            this.zkNodePaths = zkNodePaths.replaceAll("/+", "/");
        else {
            this.zkNodePaths = (ZKCONFIGS_CHROOT + "/" + zkNodePaths).replaceAll("/+", "/");
        }
    }

    private String connString;
    private String chKCmd;

    private boolean zkResouceEnable = true;
    private boolean regression;
    private OnConnectionFailed onConnectionFailed;
    private ReloadContext reloadContext;
    private RefreshContextWatcher watcher;
    private ZooKeeper zk;

    public ZookeeperResource() {
        this.zkResouceEnable = ZkResourceConstants.enable;
        this.connString = ZKClient.get().getConnAddress();
        this.regression = ZkResourceConstants.regression;
        this.onConnectionFailed = OnConnectionFailed.valueOf(ZkResourceConstants.on_connection_failed);
        this.reloadContext = ReloadContext.valueOf(ZkResourceConstants.reload_context);
    }

    public ZookeeperResource(String connString, String znodes, PingCmd chkCmd, boolean regression,
                             OnConnectionFailed onConnectionFailed, ReloadContext reloadContext) {
        this.connString = connString;
        this.zkNodePaths = znodes;
        this.chKCmd = generateZkPingCmd(chkCmd);
        this.regression = regression;
        this.reloadContext = reloadContext;
        this.onConnectionFailed = onConnectionFailed;
    }


    private String generateZkPingCmd(PingCmd pingCmd) {
        return "zkCli -server " + connString + " " + pingCmd.toString() + " " + zkNodePaths.split(",")[0];
    }


    static class StreamWriter extends Thread {
//        Logger log;

        BufferedReader br;

        StreamWriter(InputStream inputStream) {
            this.br = new BufferedReader(new InputStreamReader(inputStream));
//            this.log = log;
            start();
        }

        public void run() {
            String line = null;
            try {
                while ((line = br.readLine()) != null) {
//                    log.info(line);
                }
            } catch (IOException e) {
//                log.error("failed to log ZK process.", e);
            }
        }
    }

    private static class ZkExecutor implements Runnable, Watcher, DataMonitorListener {
//        private static Logger log = LoggerFactory.getLogger(ZkExecutor.class);

        private String cmd;
        private Process child;
        private ZooKeeper zk;
        private DataMonitor dm;
        private String znodes;
        private ZookeeperResource zkRes;

        private boolean zkResStarted = false;

        public ZkExecutor(ZookeeperResource zkRes) throws IOException {

            this.znodes = zkRes.zkNodePaths;
            this.cmd = zkRes.chKCmd;
            this.zk = ZKClient.get().getZooKeeper();
            this.dm = new DataMonitor(zk, znodes, zkRes.watcher, this);
            this.zkRes = zkRes;
        }


        @Override
        public void process(WatchedEvent event) {
            dm.process(event);
            if (!zkResStarted) {
                synchronized (zkRes) {
                    zkRes.notify();
                    zkResStarted = true;
                }
            }
        }

        @Override
        public void run() {
            try {
                synchronized (this) {
                    while (!dm.dead) {
                        wait();
                    }
                }
            } catch (InterruptedException e) {
//                log.error("InterruptedException in run", e);
            }

        }

        @Override
        public void exists(byte[] data) {
            if (data == null) {
                if (child != null) {
//                    log.info("Stopping ZK process.");
                    child.destroy();
                    try {
                        child.waitFor();
                    } catch (InterruptedException e) {
//                        log.error("Error found when waiting for ZK process stopping", e);
                    }
                }
                child = null;
            } else {
                if (child != null) {
//                    log.info("Stopping ZK process.");
                    child.destroy();
                    try {
                        child.waitFor();
                    } catch (InterruptedException e) {
//                        log.error("Error found when waiting for ZK process stopping", e);
                    }
                }
//                log.info("Load config from zookeeper:\n" + new String(data));
                try {
//                    log.info("Starting ZK process.");
                    child = Runtime.getRuntime().exec("dir");
//                    new StreamWriter(child.getInputStream(), log);
                } catch (Exception e) {
//                    log.error("Failed to start ZK process.", e);
                }
            }

        }

        @Override
        public void closing(int rc) {
            synchronized (this) {
                notifyAll();
            }

        }

        public ZooKeeper getZk() {
            return this.zk;
        }
    }

    public interface DataMonitorListener {
        void exists(byte[] data);

        void closing(int rc);
    }

    private static class DataMonitor implements Watcher, StatCallback {

        public boolean dead;
        private DataMonitorListener listener;
        private ZooKeeper zk;
        private Watcher chainedWatcher;
        private byte prevData[];

        public DataMonitor(ZooKeeper zk, String znodes, Watcher watcher, DataMonitorListener listener) {
            this.zk = zk;
            this.chainedWatcher = watcher;
            this.listener = listener;
            for (String znode : znodes.split(",")) {
                zk.exists(znode, true, this, null);
            }
        }

        @Override
        public void process(WatchedEvent event) {
            String path = event.getPath();
            if (event.getType() == Event.EventType.None) {
                switch (event.getState()) {
                    case SyncConnected:
                        break;
                    case Expired:
                        dead = true;
                        listener.closing(KeeperException.Code.SESSIONEXPIRED.intValue());
                        break;
                    default:
//                        log.info("Recevied zk change with unknow status:" + event.getState() + ", skip.", "", "");
                        break;
                }
            } else {
                if (path != null && path.equals(path)) {
                    zk.exists(path, true, this, null);
                }
            }
            if (chainedWatcher != null) {
                chainedWatcher.process(event);
            }
        }

        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            boolean exists;
            Code code = Code.get(rc);
            switch (code) {
                case OK:
                    exists = true;
                    break;
                case NONODE:
                    exists = false;
                    break;
                case SESSIONEXPIRED:
                case NOAUTH:
                    dead = true;
                    listener.closing(rc);
                    return;
                default:
                    zk.exists(path, true, this, null);
                    return;
            }

            byte b[] = null;
            if (exists) {
                try {
                    b = zk.getData(path, false, null);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    return;
                }
            }
            if ((b == null && b != prevData) || (b != null && !Arrays.equals(prevData, b))) {
                listener.exists(b);
                prevData = b;
            }
        }

    }

    @Override
    public boolean exists() {
        try {
            Stat stat = ZKClient.get().getZooKeeper().exists(zkNodePaths, false);
            return null != stat;
        } catch (Exception e) {
//            log.error("Falied to detect the config in zoo keeper.", e);
            return false;
        }
    }

    @Override
    public boolean isOpen() {
        return false;
    }

    @Override
    public URL getURL() throws IOException {
        return new URL(URL_HEADER + connString + zkNodePaths);
    }

    @Override
    public String getFilename() throws IllegalStateException {
        return zkNodePaths;
    }

    @Override
    public String getDescription() {
        return "Zookeeper resouce at '" + URL_HEADER + connString + ", zonode: '" + zkNodePaths + "'. Enabled: "
                + zkResouceEnable;
    }

    @Override
    public InputStream getInputStream() throws IOException {

        try {
            return new ByteArrayInputStream(ZKClient.get().getData(zkNodePaths));
        } catch (InterruptedException e) {
            throw new org.springframework.context.ApplicationContextException("Failed to connect to zk server"
                    + this.connString, e);
        } catch (KeeperException e) {
            throw new org.springframework.context.ApplicationContextException("Failed to connect to zk server"
                    + this.connString, e);
        }
    }


    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        this.ctx = (AbstractApplicationContext) ctx;
    }

    @Override
    public void destroy() throws Exception {

    }


}
